package com.yaoyong.yy_demo.module.rabbitmq.springboot;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.Map;

import static com.yaoyong.yy_demo.config.RabbitMQConfig.DEAD_LETTER_QUEUE;
import static com.yaoyong.yy_demo.config.RabbitMQConfig.DELAY_QUEUE;
import static com.yaoyong.yy_demo.config.RabbitMQConfig.QUEUE_A;

@Slf4j
@Component
public class Consumer {

    /**
     * 接收普通队列消息
     */
    @RabbitListener(queues = QUEUE_A)
    public void processMessage(Message message, Channel channel, @Headers Map<String, Object> headers) throws IOException {
        try {
            //每次处理一条消息
            channel.basicQos(1);
            String msg = new String(message.getBody(), "UTF-8");
            log.info("接收到普通队列的消息：---{}---消息ID---{}", msg, headers.get("spring_returned_message_correlation"));
            /**
             * 第一个参数，消息的标识
             * RabbitMQ 推送消息给 Consumer 时,会附带一个 Delivery Tag，以便 Consumer可以在消息确认时告诉 RabbitMQ 到底是哪条消息被确认了;
             * 第二个参数 multiple 取值为 false 时，表示通知 RabbitMQ 当前消息被确认；
             * 如果为 true，则额外将比第一个参数指定的 delivery tag 小的消息一并确认。
             */
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
            /**
             * 第一个参数，消息的标识
             * 第二个参数是否批量处理消息,true:将一次性拒绝所有小于deliveryTag的消息。
             * 第三个参数，消息是否重入队列，false将消息存队列删除。true:会重复消费该消息直到消息被确认消费
             */
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }

    /**
     * 接收延迟消息
     */
    @RabbitListener(queues = DELAY_QUEUE)
    public void processMessage1(Message message, Channel channel, @Headers Map<String, Object> headers) throws IOException {
        try {
            String msg = new String(message.getBody(), "UTF-8");
            log.info("接收到延迟消息:---{}---消息ID---{}", msg, headers.get("spring_returned_message_correlation"));
        } catch (UnsupportedEncodingException e) {
            //第二个false将消息存队列删除,true放回队列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    /**
     * 接收死信队列消息
     */
    @RabbitListener(queues = DEAD_LETTER_QUEUE)
    public void processMessage2(Message message, Channel channel, @Headers Map<String, Object> headers) throws IOException {
        try {
            //可以考虑数据库记录，每次进来查数量，达到一定的数量，进行预警，人工介入处理
            String msg = new String(message.getBody(), "UTF-8");
            log.info("接收到死信消息:---{}---消息ID---{}", msg, headers.get("spring_returned_message_correlation"));
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        //回复ack
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}
